package com.jumei.flume.handler.shuabao;

import com.google.common.base.Function;
import com.google.common.collect.Lists;
import com.jumei.flume.bo.ShuabaoSensorMessage;
import com.jumei.flume.filter.EventsFilter;
import com.jumei.flume.handler.Handler;
import com.jumei.flume.parser.JSONStringToShuabaoSensorMessageConverter;
import com.jumei.flume.sink.kudu.KuduConfiguration;
import com.jumei.flume.sink.kudu.KuduJdbcDao;
import com.jumei.flume.sink.kudu.KuduJdbcDaoImpl;
import org.apache.flume.Event;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.List;

public class ShuabaoSensorKuduJdbcHandler implements Handler {
	private final static Logger LOGGER = LoggerFactory.getLogger(ShuabaoSensorKuduJdbcHandler.class);

	KuduJdbcDao dao4shuabao;
	
	@Override
	public void start() {
		//kudu连接初始化
		KuduConfiguration c = new KuduConfiguration();
		if(null == dao4shuabao) {
			dao4shuabao = new KuduJdbcDaoImpl(c.kuduClient4Shuabao());
		}
	}

	@Override
	public void stop() {
		dao4shuabao = null;
	}

	static JSONStringToShuabaoSensorMessageConverter converter = new JSONStringToShuabaoSensorMessageConverter();
	
	
	@Override
	public void handler(List<Event> eventList) throws Exception {
		
		if(CollectionUtils.isEmpty(eventList)) {
			return;
		}

		List<ShuabaoSensorMessage> records = Lists.transform(eventList, new Function<Event, ShuabaoSensorMessage>() {

			@Override
			public ShuabaoSensorMessage apply(Event event) {
				try {
					String line = new String(event.getBody());
					
					ShuabaoSensorMessage sensorMessage = converter.convert(line);
		            if (null == sensorMessage) {
		                return null;
		            }
		            
		            //只选择指定项目的指定事件
		            if (!EventsFilter.doFilter4Shuabao(sensorMessage)) {
		                return null;
		            }


					System.out.println(sensorMessage.toString());
					return sensorMessage;
				} catch (Exception e) {
					//say something here
					LOGGER.error("handler error: ", e);
				}
				
				return null;
			}
			
		});
//
//		int kuduUpsertNumer4tx = dao4shuabao.upsert(records);
//		LOGGER.info("Upsert into shuabao's kudu records: {}, to tx {}.", 0, kuduUpsertNumer4tx);
	}


}
